package cn.mesmile.kafka.controller;

import lombok.AllArgsConstructor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.kafka.core.KafkaOperations;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

/**
 * @author zb
 * @date 2021/10/31 16:47
 * @Description
 */
@RequestMapping("/api/v1/kafka")
@RestController
@AllArgsConstructor
public class UserController {

    private static final String TOPIC_NAME = "hello_kafka";

    private final KafkaTemplate<String, Object> kafkaTemplate;


    @GetMapping("/send/{num}")
    public Object sendMessage(@PathVariable("num")Integer num){
        kafkaTemplate.send(TOPIC_NAME,"这是一条springboot发送的消息，消息序列号为："+num).addCallback(
                success->{
                    RecordMetadata recordMetadata = success.getRecordMetadata();
                    ProducerRecord<String, Object> producerRecord = success.getProducerRecord();

                    System.out.println("recordMetadata.offset() = " + recordMetadata.offset());
                    System.out.println("recordMetadata.partition() = " + recordMetadata.partition());
                    System.out.println("recordMetadata.topic() = " + recordMetadata.topic());

                },fail->{
                    System.out.println("fail.getMessage() = " + fail.getMessage());

                });
        return "success";
    }

    /**
     * 注解方式的事务
     * @param num
     */
    @GetMapping("/transaction1/{num}")
    @Transactional(rollbackFor = RuntimeException.class)
    public void sendMessage1(@PathVariable("num")Integer num) {
        kafkaTemplate.send(TOPIC_NAME, "这个是事务里面的消息：1  num="+num);
        if (num == 0) {
            throw new RuntimeException("fail");
        }
        kafkaTemplate.send(TOPIC_NAME, "这个是事务里面的消息：2  num="+num);
    }
    /**
     * 声明式事务支持
     * @param num
     */
    @GetMapping("/transaction2/{num}")
    public void sendMessage2(@PathVariable("num")Integer num) {
        kafkaTemplate.executeInTransaction(new KafkaOperations.OperationsCallback() {
            @Override
            public Object doInOperations(KafkaOperations kafkaOperations) {
                kafkaOperations.send(TOPIC_NAME,"这个是事务里面的消息：1  num="+num);
                if(num==0){
                    throw new RuntimeException("input is error");
                }
                kafkaOperations.send(TOPIC_NAME,"这个是事务里面的消息：2  num="+num);
                return true;
            }
        });
    }


}
